package org.example.service;

import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.lang.Console;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.grpc.stub.StreamObserver;
import lombok.SneakyThrows;
import org.example.grpc.TestFirstServiceGrpc;
import org.example.model.TestProto;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

/**
 * 实现服务具体的功能接口方法
 *
 * @author LinRuChang
 * @version 1.0
 * @date 2025/10/13
 * @since 1.8
 **/
public class TestFirstService extends TestFirstServiceGrpc.TestFirstServiceImplBase {

    // 使用jackson进行转成 json字符串
    static ObjectMapper mapper = new ObjectMapper();

    static {
        mapper.findAndRegisterModules();
    }

    @Override
    @SneakyThrows
    public void helloWorld(TestProto.RequestData request, StreamObserver<TestProto.ResponseData> responseObserver) {

        // 1. 接收参数信息
        Console.log("【helloWorld】接收到的请求参数：{}", mapper.writeValueAsString(request));

        // 2. 业务处理
        TestProto.ResponseData responseData = TestProto.ResponseData.newBuilder()
                .setUserInfo(TestProto.RequestData.UserInfo.newBuilder()
                        .setName("linrc")
                        .build())
                .setSuccessFlag(true)
                .build();

        // 3. 返回结果给调用方
        responseObserver.onNext(responseData);

        // 4. 请求结束
        responseObserver.onCompleted();
    }


    @Override
    @SneakyThrows
    public void helloWorldBySSE(TestProto.RequestData request, StreamObserver<TestProto.ResponseData> responseObserver) {
        TimeInterval timeInterval = new TimeInterval();

        // 1. 接收参数信息
        Console.log("【helloWorldBySSE】接收到的请求参数：{}", mapper.writeValueAsString(request));

        // 2. 业务处理
        int randomCount = RandomUtil.randomInt(3, 10);
        Stream.iterate(0, i -> i + 1)
                .limit(randomCount)
                .forEach(i -> {

                    TestProto.ResponseData responseData = TestProto.ResponseData.newBuilder()
                            .setUserInfo(TestProto.RequestData.UserInfo.newBuilder()
                                    .setName(StrUtil.format("linrc_{}_{}", i, IdUtil.fastSimpleUUID()))
                                    .build())
                            .setSuccessFlag(true)
                            .build();
                    ThreadUtil.sleep(1, TimeUnit.SECONDS);


                    // 3. 返回结果给调用方
                    responseObserver.onNext(responseData);

                });

        // 4. 请求结束
        Console.log("【helloWorldBySSE】接收到的请求结束：发送消息{}个 == 共耗时{}ms", randomCount, timeInterval.intervalRestart());
        responseObserver.onCompleted();
    }


    @Override
    public StreamObserver<TestProto.RequestData> helloWorldByClientStreamRpc(StreamObserver<TestProto.ResponseData> responseObserver) {

        AtomicInteger messageCount = new AtomicInteger();
        StreamObserver<TestProto.RequestData> result = new StreamObserver<>() {

            @Override
            @SneakyThrows
            public void onNext(TestProto.RequestData request) {
                //可以这里每接收到一次请求消息，就处理 == 看业务需求
                Console.log("【helloWorldByClientStreamRpc】接收到的请求参数消息【{}】：{}", messageCount.incrementAndGet(), mapper.writeValueAsString(request));
            }

            @Override
            public void onError(Throwable e) {
                Console.log("【helloWorldByClientStreamRpc】所有消息异常", e);
            }

            @Override
            public void onCompleted() {
                //可以这里处理接收到的所有请求消息，在进行处理 == 看业务需求
                Console.log("【helloWorldByClientStreamRpc】所有消息接收完毕：总请求消息个数【{}】", messageCount.get());

                // 提供服务端响应给客户端
                TestProto.ResponseData responseData = TestProto.ResponseData.newBuilder()
                        .setSuccessFlag(true)
                        .setRemark(StrUtil.format("服务端接收到的消息成功，共【{}】条", messageCount.get()))
                        .build();
                responseObserver.onNext(responseData);
                responseObserver.onCompleted();

            }
        };

        return result;
    }

    @Override
    public StreamObserver<TestProto.RequestData> helloWorldByWebSocket(StreamObserver<TestProto.ResponseData> responseObserver) {

        AtomicInteger messageCount = new AtomicInteger();
        StreamObserver<TestProto.RequestData> requestDataStreamObserver = new StreamObserver<>() {

            @Override
            @SneakyThrows
            public void onNext(TestProto.RequestData request) {
                Console.log("【helloWorldByWebSocket】接收到的请求参数消息【{}】：{}", messageCount.incrementAndGet(), mapper.writeValueAsString(request));

                // 提供服务端响应给客户端
                TestProto.ResponseData responseData = TestProto.ResponseData.newBuilder()
                        .setSuccessFlag(true)
                        .setRemark(StrUtil.format("服务端已接收到的消息【{}】", messageCount.get()))
                        .build();
                responseObserver.onNext(responseData);
                // responseObserver.onCompleted();

            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onCompleted() {
                //可以这里处理接收到的所有请求消息，在进行处理 == 看业务需求
                Console.log("【helloWorldByWebSocket】所有消息接收完毕：总请求消息个数【{}】", messageCount.get());
                responseObserver.onCompleted();
            }
        };



        return requestDataStreamObserver;
    }
}
